import type { Logger } from "@smithy/types";

import { ByteArrayCollector } from "./ByteArrayCollector";
import type { BufferStore, Modes } from "./createBufferedReadableStream";
import { flush, merge, modeOf, sizeOf } from "./createBufferedReadableStream";
import { buffer, stream } from "@kit.ArkTS";
import hilog from "@ohos.hilog";
import { PassThrough } from "./passThrough";

/**
 * @internal
 * @param upstream - any Readable or ReadableStream.
 * @param size - byte or character length minimum. Buffering occurs when a chunk fails to meet this value.
 * @param onBuffer - for emitting warnings when buffering occurs.
 * @returns another stream of the same data and stream class, but buffers chunks until
 * the minimum size is met, except for the last chunk.
 */
export function createBufferedReadable(upstream: stream.Readable, size: number, logger?: Logger): stream.Readable {
  return new BufferedReadable(upstream,size,logger)
}

const TAG = 'BufferedReadable'
export class BufferedReadable extends PassThrough{
  private streamBufferingLoggedWarning = false;
  private bytesSeen = 0;
  private buffers = [
    "",
    new ByteArrayCollector((size) => new Uint8Array(size)),
    new ByteArrayCollector((size) => new Uint8Array(size))
  ] as BufferStore;
  private mode: Modes | -1 = -1;
  constructor(private upstream: stream.Readable,private  size: number,private  logger?: Logger) {
    super(upstream,TAG);
    this.upstream.pipe(this);
  }

  doPush(chunk: string | Uint8Array, encoding: string, callback: Function): void {
    hilog.warn(0,TAG,'do push')
    const chunkMode = modeOf(chunk, true);
    if (this.mode !== chunkMode) {
      if (this.mode >= 0) {
        this.push(flush(this.buffers, this.mode));
      }
      this.mode = chunkMode;
    }
    if (this.mode === -1) {
      this.push(chunk);
      callback()
      return;
    }

    const chunkSize = sizeOf(chunk);
    this.bytesSeen += chunkSize;
    const bufferSize = sizeOf(this.buffers[this.mode]);
    if (chunkSize >= this.size && bufferSize === 0) {
      // skip writing to the intermediate buffer
      // because the upstream chunk is already large enough.
      this.push(chunk);
    } else {
      // buffer and potentially flush the data downstream.
      const newSize = merge(this.buffers, this.mode, chunk);
      if (!this.streamBufferingLoggedWarning && this.bytesSeen > this.size * 2) {
        this.streamBufferingLoggedWarning = true;
        this.logger?.warn(
          `@smithy/util-stream - stream chunk size ${chunkSize} is below threshold of ${this.size}, automatically buffering.`
        );
      }
      if (newSize >= this.size) {
        this.push(flush(this.buffers, this.mode));
      }
    }
    callback()
  }

  async finally(): Promise<void> {
    if (this.mode !== -1) {
      const remainder = flush(this.buffers, this.mode);
      if (sizeOf(remainder) > 0) {
        this.push(remainder);
      }
    }
  }
}
